from sklearn.metrics import roc_auc_score, confusion_matrix, ConfusionMatrixDisplay, RocCurveDisplay, brier_score_loss
from sklearn.calibration import CalibrationDisplay
from joblib import dump, load
from skopt import BayesSearchCV
from skopt.space import Real, Categorical, Integer
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import RobustScaler, OrdinalEncoder, OneHotEncoder
import shap
import pprint
from IPython.display import HTML, display
import tabulate
class MyPipeline(Pipeline):
# https://stackoverflow.com/questions/33469633/how-to-transform-items-using-sklearn-pipeline
def just_transforms(self, X):
"""Applies all transforms to the data, without applying last
estimator.
Parameters
----------
X : iterable
Data to predict on. Must fulfill input requirements of first step of
the pipeline.
"""
Xt = X
for name, transform in self.steps[:-1]:
Xt = transform.transform(Xt)
return self.steps[-1][1], Xt
def dict_to_table(d, header=['', '']):
arr = [[a,b] for a,b in d.items()]
table = tabulate.tabulate(arr, tablefmt='html', headers=header)
display(HTML(table))
def variables_olas(olas):
olas_str = [str(x) for x in olas]
variables = set()
#variables = variables.union(set(f_sel['rf_' + '.'.join(olas_str)]['EXITUS']))
variables = variables.union(set(f_sel['gbm_' + '.'.join(olas_str)]['EXITUS']))
variables = [translate[v] for v in variables]
return list(variables)
def seleccion_olas(df, olas):
variables = variables_olas(olas)
return df[df[translate['Ola']].isin(olas)][variables + [translate['EXITUS']]]
def model_tune(model, tune_grid, X_train, y_train, scale=False, encode=None, n_iter=50, verbose=0):
steps = []
if encode == 'ordinal':
steps.append(('encode', OrdinalEncoder()))
elif encode == 'ohe':
steps.append(('encode', OneHotEncoder()))
if scale:
steps.append(('scale', RobustScaler()))
if steps:
steps.append(('model', model))
model = MyPipeline(steps)
tune_grid = {f'model__{p}': tune_grid[p] for p in tune_grid.keys()}
tuned_model = BayesSearchCV(
model, tune_grid, n_jobs=-1, cv=5, random_state=SEED, scoring='roc_auc', n_iter=n_iter, verbose=verbose
).fit(X_train, y_train)
print(f'Tune result: {tuned_model.best_score_} auc.')
return tuned_model.best_estimator_, tuned_model.best_score_
def model_report(model, X, y, id_persistencia=None):
out = {'metrics': {}, 'model': model, 'y': y}
feature_names = list(X.columns)
out['feature_names'] = feature_names
if str(type(model)) == str(MyPipeline):
model, X = model.just_transforms(X)
metrics = out['metrics']
out['y_prob'] = y_prob = model.predict_proba(X)[:, 1]
metrics['roc'] = roc_auc_score(y, y_prob)
metrics['brier_loss'] = brier_score_loss(y, y_prob, pos_label="S")
out['y_pred'] = y_pred = model.predict(X)
tn, fp, fn, tp = confusion_matrix(y, y_pred).ravel()
metrics['specificity'] = tn / (tn + fp)
metrics['sensitivity'] = tp / (tp + fn)
metrics['accuracy'] = (tp + tn) / (tp + tn + fp + fn)
if id_persistencia:
dump(out, id_persistencia)
return out
def shap_plot(report, X_train, X_val, shap_explainer=None):
if shap_explainer:
model = report['model']
if str(type(model)) == str(MyPipeline):
_, X_train = model.just_transforms(X_train)
model, X_val = model.just_transforms(X_val)
if shap_explainer == shap.TreeExplainer:
shap_values = shap_explainer(model, X_train, seed=SEED).shap_values(X_train, check_additivity=False)
else:
shap_values = shap_explainer(model, X_train, seed=SEED).shap_values(X_train)
report['shap'] = shap_values
_ = plt.figure(figsize=(15,6))
if type(shap_values) == type([]):
shap_values = shap_values[1]
shap.summary_plot(shap_values, X_train, report['feature_names'], show=False, plot_size=None)
plt.show()
def plot_report(report, model_repr=None):
fig, axs = plt.subplot_mosaic([['left', 'right'],['bottom', 'bottom']],
figsize=(10, 10), constrained_layout=True)
RocCurveDisplay.from_predictions(report['y'], report['y_prob'], pos_label='S', ax=axs['left'])
axs['left'].set_xticks(np.arange(0, 1, step=0.1))
axs['left'].set_yticks(np.arange(0, 1, step=0.1))
CalibrationDisplay.from_predictions(report['y'], report['y_prob'], pos_label='S', ax=axs['right'])
axs['right'].set_xticks(np.arange(0, 1, step=0.1))
axs['right'].set_yticks(np.arange(0, 1, step=0.1))
ConfusionMatrixDisplay.from_predictions(report['y'], report['y_pred'], ax=axs['bottom'])
plt.show()
dict_to_table(report['metrics'], ['Metric', 'Value'])
model = report['model']
if model_repr:
if str(type(model)) == str(MyPipeline):
model = model.steps[-1][1]
model_repr(model, report['feature_names'])
dict_to_table(model.get_params(), ['Hyperparameter', 'Value'])
def generar_id(modelo, olas, disc=False, es_test=False):
if modelo is None:
return None
return ('modelosPersistidos/' +
modelo +
'_'.join([str(o) for o in olas]) +
('_disc' if disc else '') +
('_test' if es_test else '') +
'.joblib')
def model_demonstrate(olas, model, tune_grid, df_train, df_val, scale=False, encode=None,
shap_explainer=None, model_repr=None, class_name='EXITUS', nombre_modelo=None,
n_iter=50, verbose=0, es_test=False):
df_train_ola = seleccion_olas(df_train, olas)
df_val_ola = seleccion_olas(df_val, olas)
class_name = translate[class_name]
predictors_ola = [x for x in df_train_ola.columns if x != class_name]
X_train, y_train = df_train_ola[predictors_ola], df_train_ola[class_name]
X_val, y_val = df_val_ola[predictors_ola], df_val_ola[class_name]
if nombre_modelo is None:
nombre_modelo = model.__class__.__name__
if n_iter > 0:
print(f'Tuning params {", ".join(list(tune_grid.keys()))}')
tuned_model, train_score = model_tune(model, tune_grid, X_train, y_train, scale=scale, n_iter=n_iter, verbose=verbose)
else:
tuned_model, train_score = model.fit(X_train, y_train), None
id_modelo = generar_id(nombre_modelo, olas, disc=(encode is not None), es_test=es_test)
report = model_report(
tuned_model, X_val, y_val, id_persistencia=id_modelo)
shap_plot(report, X_train, X_val, shap_explainer)
report['train_score'] = train_score
plot_report(report, model_repr=model_repr)
return report, (X_train, y_train, X_val, y_val)
